/*
 * XCDBSource.java
 *
 * Created on August 29, 2005, 8:53 PM
 *
 * To change this basis, choose Tools | Options and locate the basis under
 * the Source Creation and Management node. Right-click the basis and choose
 * Open. You can then make changes to the basis in the Source Editor.
 */
package com.xucia.jsponic.datasource;
import java.security.acl.Acl;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.json.JSONObject;

import com.xucia.jsponic.data.DataSourceHelper;
import com.xucia.jsponic.data.DataSourceManager;
import com.xucia.jsponic.data.GlobalData;
import com.xucia.jsponic.data.ObjectId;
import com.xucia.jsponic.data.Persistable;
import com.xucia.jsponic.data.PersistableImpl;
import com.xucia.jsponic.data.PersistableList;
import com.xucia.jsponic.data.JsonPath;
import com.xucia.jsponic.data.Stipulation;
import com.xucia.jsponic.data.Templates;
import com.xucia.jsponic.data.LazyPropertyId;
import com.xucia.jsponic.query.CombinationCondition;
import com.xucia.jsponic.query.ComparisonCondition;
import com.xucia.jsponic.query.Condition;
import com.xucia.jsponic.query.ConstantExpression;
import com.xucia.jsponic.query.Expression;
import com.xucia.jsponic.query.IdentifierExpression;
import com.xucia.jsponic.security.ObjectAccessDeniedException;
import com.xucia.jsponic.security.PersistentAcl;

/*

CREATE TABLE str (
id INT NOT NULL AUTO_INCREMENT,
value MEDIUMTEXT,
PRIMARY KEY(id)
)

CREATE TABLE obj (
id INT NOT NULL,
field INT NOT NULL,
value_type TINYINT(4) NOT NULL,
value INT NOT NULL,
PRIMARY KEY(id, field)
)

CREATE TABLE list (
list_id INT NOT NULL,
row_id INT NOT NULL AUTO_INCREMENT,
value_type TINYINT(4) NOT NULL,
value INT NOT NULL,
PRIMARY KEY(row_id),INDEX (list_id)
)
 */ 
public class DynaObjectDBSource extends DatabaseDataSource implements QueriableDataSource, IndexedListDataSource, WritableDataSource, DataSource, DataSourceCanHaveOrphans, ReferenceAwareDataSource {

	public void mapQuery(PersistableInitializer initializer, JsonPath query)
			throws Exception {
		StringBuffer sql = new StringBuffer("SELECT value_type,value FROM list WHERE list_id=" + query.getFrom().subObjectId + " AND value IN (SELECT id FROM obj WHERE ");
        addConditionToQuery(((JsonPath) query).getWhere(), sql);
        sql.append(')');
        //System.err.println("Sublist query: " + sql);
    	initializeList(initializer,executeQuery(sql.toString()));
	}
	public DynaObjectDBSource() {
		
	}
    @Override
	public void initParameters(JSONObject parameters) throws Exception {
    	super.initParameters(parameters);
    	findMaxId();
	}
	public List<ObjectId> getOrphans() {
    	throw new UnsupportedOperationException("not implemented yet");    	
	}
    Set<String> marked;
    Set<Persistable> hasPermission = new HashSet();

    private void mark(final ObjectId object, final ObjectId prototype) {
    	//TODO: We need to be able to remove entries from alterations that are alterations of unreachable objects
		final boolean newlyVisited = marked.add(object.subObjectId);
		try {
			Long.parseLong(object.subObjectId); // throw a parse exception right away if it is not a number
			mapObject(new PersistableInitializer() {
				boolean prototypeExists = false;
				public void setSchema(Persistable structure) {}


				public void setCacheLevel(long time) {
				}

				public void initializeByPrototype(ObjectId prototypeId) {
					setProperty(GlobalData.BASIS_FIELD, prototypeId);
				}

				public void finished() {
					if (!prototypeExists && prototype != null)
						try {
							recordPropertyAddition(object.subObjectId, "prototype", prototype);
						} catch (SQLException e) {
							
						}
				}

				public Persistable getInitializingObject() {
					return null;
				}

				public void initializeFunction(String source, boolean authorizedForServer) {
				}

				public void initializeList(Iterator iterator)  {
					while (iterator.hasNext()) {
						Object value = iterator.next();
						if (newlyVisited && value instanceof ObjectId)
							mark((ObjectId) value,null);
					}
				}

				public void initializeObject(PersistableImpl object) {
				}

				public void setProperty(String key, Object value) {
					try {
					if (newlyVisited && value instanceof ObjectId) {
						if ("prototype".equals(key) && prototype != null)
							prototypeExists = true;
						if ("history".equals(key))
						{}
						else if (GlobalData.PARENT_FIELD.equals(key))
						{}//	mark((DataObject) value,object);
						else
							mark((ObjectId) value,"structure".equals(key) || "constructor".equals(key) ? object:null);
					}
					else if (value instanceof ShortStringLookupPair) {
				        ResultSet rs = null;
				        PreparedStatement statement = getConnectionObject().stringLookup;
				    	statement.setLong(1,Long.parseLong(((LazyPropertyId)value).subObjectId));
				        rs = statement.executeQuery();
				        if (rs.next()) 
				            if (decodeString(rs.getString("value")).indexOf("mthd") != -1)
				            	recordPropertyRemoval(object.subObjectId,key);
				        rs.close();
					}
					}
					catch(NumberFormatException e) {    				
					}
					catch(Exception e) {
						e.printStackTrace();
					}
				}

				public void setAcl(Acl acl) {}
				public void setPersistentAcl(ObjectId aclId) {}
				public void setParent(ObjectId objectToInheritFrom) {}
			},object.subObjectId);

		} catch (NumberFormatException e) {
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
    }

	public synchronized void purgeOrphans() {
//		 TODO: This is using mark and sweep, but it is not safe while other operations are happening, 
		//therefore now it can only be run on startup, but it needs to be implemented using a reference tracing algorithm so it can run concurrently
		
		try {
			marked = new HashSet();
			mark(ObjectId.idForObject(DataSourceManager.getRootSource(),"0"),null);		
			ResultSet rs = executeQuery("SELECT DISTINCT id FROM obj");
			while (rs.next()) {
				long id = rs.getInt("id");
				if (!marked.contains("" + id)) { 
//TODO: This should be enabled if running in runtime:					if (sourceObjectMap.get(id) == null) {// It has to be garbage collected by Java as well, so we don't destroy an object that is newly created or being transferred
						System.err.println("Deleting " + id);
						ResultSet parentRs = executeQuery("SELECT value FROM obj WHERE field='parent' AND id = " + id);
						try {
							parentRs.next();
							long newParentId = parentRs.getLong(1);
							executeQuery("UPDATE obj SET value=" + newParentId + " WHERE field='parent' AND value = " + id);
						}
						catch (SQLException e) {
							e.printStackTrace();
						}
						executeQuery("DELETE FROM obj WHERE id = " + id).close();
					}
				//}
			}
			rs = executeQuery("SELECT DISTINCT list_id FROM list");
			while (rs.next()) {
				long id = rs.getInt("list_id");
				if (!marked.contains("" + id)) {
//					if (sourceObjectMap.get(id) == null) {// It has to be garbage collected by Java as well, so we don't destroy an object that is newly created or being transferred
						System.err.println("Deleting " + id);
						executeQuery("DELETE FROM list WHERE list_id = " + id).close();
	//				}
				}
			}
			rs.close();
			rs = executeQuery("DELETE FROM str WHERE id NOT IN (SELECT value FROM obj WHERE value_type=" + STR + " UNION SELECT value FROM list WHERE value_type=" + STR + ")");
			rs.close();
			marked = null;
		//	fixPermission();
		}
		catch (SQLException e) {
			throw new RuntimeException(e);
		}
		
	}

	static {  
        try { 
            //Class.forName("com.mysql.jdbc.Driver").newInstance();
            Class.forName("org.hsqldb.jdbcDriver").newInstance();
        } catch (Exception ex) {
            ex.printStackTrace(); 
        }
    }
    ResultSet rs = null;
    @Override
	ThreadSpecificConnectionObject setupConnection(Connection connection) throws SQLException {
    	ConnectionStatements statements = new ConnectionStatements();
    	statements.objectRetrieval = connection.prepareStatement("SELECT field,value_type,value FROM obj where id=?");
    	statements.referrerLookup = connection.prepareStatement("SELECT id FROM obj where value=? AND value_type=" + OBJECT_TYPE + " AND NOT field='parent' UNION SELECT list_id as id FROM LIST WHERE value=? AND value_type=" + OBJECT_TYPE);
    	statements.allReferrerLookup = connection.prepareStatement("SELECT id FROM obj where value=? AND value_type=" + OBJECT_TYPE + " UNION SELECT list_id as id FROM LIST WHERE value=? AND value_type=" + OBJECT_TYPE);
    	statements.stringLookup = connection.prepareStatement("SELECT value FROM str where id=?");
    	statements.listRetrieval = connection.prepareStatement("SELECT value_type,value FROM list WHERE list_id=?");
    	statements.selectRowId = connection.prepareStatement("SELECT row_id FROM list WHERE list_id=?");
    	statements.insertRow = connection.prepareStatement("INSERT INTO list (list_id,value_type,value) values (?,?,?)");
		if (connection instanceof org.hsqldb.jdbc.jdbcConnection) 
			statements.stringCreation = connection.prepareStatement("INSERT INTO str (value) values (?)");
		else
			statements.stringCreation = connection.prepareStatement("INSERT INTO str (value) values (?)",Statement.RETURN_GENERATED_KEYS);
    	return statements;
	}
    @Override
    protected ConnectionStatements getConnectionObject() {
    	return (ConnectionStatements) super.getConnectionObject();
    }
    class ConnectionStatements extends ThreadSpecificConnectionObject{
    	public PreparedStatement objectRetrieval;
    	public PreparedStatement referrerLookup;
    	public PreparedStatement allReferrerLookup;
    	public PreparedStatement stringLookup;     	
    	public PreparedStatement stringCreation;     	
    	public PreparedStatement listRetrieval;     	
    	public PreparedStatement selectRowId;     	
    	public PreparedStatement insertRow;     	
    }
    String clientSession;
    /** id 0 is the root object
     * id 1 is the super field object
     */
    long currentId = 2;  // the id for the next object to be created
    public synchronized long getNextId()  {
        return currentId++;
    }
    void findMaxId() throws SQLException {
        rs = executeQuery("SELECT MAX(id) as mx FROM obj");
        if (rs.next())
            currentId = rs.getLong(1);
        rs.close();
        rs = executeQuery("SELECT MAX(value) as mx FROM obj WHERE value_type=" + OBJECT_TYPE);
        if (rs.next()) {
            long highPointer = rs.getLong(1);
            if (highPointer > currentId)
                currentId = highPointer;
        }
        currentId++;
        rs.close();
    }
    void establishTables() throws SQLException {
        try {
            findMaxId();
        }
        catch (SQLException e)
        {
            e.printStackTrace();
/*            System.err.println("creating the tables");
            // need to build the tables
            DataObject obj = (DataObject) rootData.get("system");
            obj = (DataObject) obj.get("XCDBDataSource");
            ListDataObjectCapable commands = (ListDataObjectCapable) obj.get("setupCommands");
            for (Object command : commands) {
                execute((String)command);
            }*/
        }
    }

    long createString(String value) throws SQLException{
    	PreparedStatement statement = getConnectionObject().stringCreation;
    	statement.setString(1,value);
    	return executeAndGetGeneratedKey(statement);
    }


    Map<Thread,Statement> threadStatements = new HashMap();
    int executeUpdate(final String sql) throws SQLException {
    	return (Integer) tryExecution(new DatabaseAction() {
			public Object execute() throws SQLException {
				Statement statement = getConnectionObject().getConnection().createStatement();
				return statement.executeUpdate(sql);
			}
    	});
    	
    }

    int executeUpdate(final String sql, final int autoGen) throws SQLException {
    	return (Integer) tryExecution(new DatabaseAction() {
			public Object execute() throws SQLException {
				Statement statement = getConnectionObject().getConnection().createStatement();
				return statement.executeUpdate(sql,autoGen);
			}
    	});
    }
    ResultSet executeQuery(final String sql) throws SQLException {
    	return (ResultSet) tryExecution(new DatabaseAction() {
			public Object execute() throws SQLException {
				Statement statement = getConnectionObject().getConnection().createStatement();
				return statement.executeQuery(sql);
			}
    	});
    }
    boolean execute(final String sql) throws SQLException {
    	return (Boolean) tryExecution(new DatabaseAction() {
			public Object execute() throws SQLException {
				Statement statement = getConnectionObject().getConnection().createStatement();
				return statement.execute(sql);
			}
    	});
    	
    }  
    public void commitTransaction() throws SQLException {
        execute("COMMIT");
    }   

    public void startTransaction() throws SQLException {
        execute("START TRANSACTION");
    }

    public void abortTransaction() throws SQLException {
        execute("ROLLBACK");
    }
    final static int NULL_TYPE = 0;
    final static int OBJECT_TYPE = 1;
    final static int BOOLEAN_TYPE = 3;
    final static int INTEGER_TYPE = 4;
    final static int STR = 5;
    final static int LONG_STR = 6;
    final static int BIG_NUM = 7;
    static class ShortStringLookupPair extends LazyPropertyId {
        ShortStringLookupPair(DataSource source, long valueId) {
            super(source, Long.toString(valueId),null);
        }
      }
      static class BigNumLookupPair extends LazyPropertyId {
        BigNumLookupPair(DataSource source, long valueId) {
            super(source, Long.toString(valueId),null);
        }
      }
      private Object getValueFromRS(ResultSet rs) throws SQLException {
          int type = rs.getInt("value_type");  

          long value = rs.getLong("value");
          switch (type) {
              case NULL_TYPE : return null;
              case OBJECT_TYPE : return ObjectId.idForObject(this,Long.toString(value));
              case BOOLEAN_TYPE : return (value != 0) ? Boolean.TRUE : Boolean.FALSE;
              case INTEGER_TYPE : return value;
              case STR : return new ShortStringLookupPair(this,value);
              case BIG_NUM : return new BigNumLookupPair(this,value);
          }
    	  throw new RuntimeException("Unknown value type");
      }
      private void initializeList(PersistableInitializer initializer, final ResultSet rs) throws SQLException {
      	initializer.initializeList(new Iterator() {
    		short fetched = 0;
			public boolean hasNext() {
				try {
					if (fetched == 0) {
						fetched = (short) (rs.next() ? 1 : -1);
						if (fetched == -1)
							rs.close();
					}
					return fetched == 1;
				} catch (SQLException e) {
					throw new RuntimeException(e);
				}
			}

			public Object next() {
				try {
					if (fetched == 0) {
						if (!rs.next()) {
							rs.close();
							fetched = -1;
						}
					}
					else if (fetched == 1)
						fetched = 0;
					else if (fetched == -1)
						throw new RuntimeException("Called next past the last item in the result set");
					return getValueFromRS(rs);
				} catch (SQLException e) {
					throw new RuntimeException(e);
				}
			}

			public void remove() {
				throw new UnsupportedOperationException();					
			}
    		
    	});
      }
      Map<String,Object> getObjectMap(String objectId) throws SQLException {
  		ResultSet rs;
        PreparedStatement statement = getConnectionObject().objectRetrieval;
        if ("".equals(objectId))
        	objectId = "0";
    	statement.setLong(1,Long.parseLong(objectId));
        rs = statement.executeQuery();
        Map<String,Object> objectMap = new HashMap(4);
        String field;
        while (rs.next()) {
            field = rs.getString("field");
            objectMap.put(field, getValueFromRS(rs));
        }
        rs.close();
        return objectMap;
      }
	public void mapObject(PersistableInitializer initializer, String objectId) throws Exception  {
		JsonPath query = parseIfQuery(objectId);
		if (query == null) {
	        Map<String,Object> objectMap = getObjectMap(objectId); 
	        Class<? extends Persistable> clazz = Templates.getClassForObject(objectMap.get(GlobalData.BASIS_FIELD));        
	        if (clazz != null)
	        	initializer.initializeObject((PersistableImpl) clazz.newInstance());
	        if (Boolean.TRUE.equals(objectMap.get(GlobalData.CONTAINS_REPEATING_IDENTIFIER))) {
	        	PreparedStatement listRetrieval = getConnectionObject().listRetrieval;
	        	listRetrieval.setLong(1, Long.parseLong(objectId));
	        	initializeList(initializer,listRetrieval.executeQuery());
	        	objectMap.remove(GlobalData.CONTAINS_REPEATING_IDENTIFIER);
	        }
	        initializeFromMap(initializer, objectMap);
	
	        Object access = objectMap.get(GlobalData.ACCESS_FIELD);
	        if (access == null) {
	        	if (objectMap.containsKey(GlobalData.PARENT_FIELD)) {
	        		if (!(objectMap.get(GlobalData.PARENT_FIELD) instanceof ObjectId))
	        			initializer.setParent((ObjectId) DataSourceManager.getRootObject().getId());
	        		else
	        			initializer.setParent((ObjectId) objectMap.get(GlobalData.PARENT_FIELD));
	        	}
	        	else {
		        	List<ObjectId> referrers = getReferrers(objectId);
		        	if (!referrers.isEmpty())
		        		initializer.setParent(referrers.get(0));
	        	}
	        }
	        else {
	        	initializer.setAcl(new PersistentAcl(((ObjectId)access).getTarget()));
	        }
		}
		else { // if it is a query
			mapQuery(initializer, query);
		}
        initializer.finished();
	}
    void addExpressionToQuery(Expression expression, final StringBuffer sql) throws SQLException {
    	if (expression instanceof IdentifierExpression) {
    		
    	}
    }
    void addConditionToQuery(Condition condition, final StringBuffer sql) throws SQLException {
        sql.append('(');
        if (condition instanceof CombinationCondition) {
        	addConditionToQuery(((CombinationCondition) condition).getCond1(), sql);
        	sql.append(" AND "); // TODO: add alternate operators
        	addConditionToQuery(((CombinationCondition) condition).getCond2(), sql);
        }
        if (condition instanceof ComparisonCondition) {
        	String name= ((IdentifierExpression) ((ComparisonCondition)condition).getLeft()).getName(); 

            sql.append("field='" + makeStringSQLSafe(name) + "'");
            Object value = ((ConstantExpression) ((ComparisonCondition)condition).getRight()).getValue();
            if (value instanceof String) {
                String valueString = (String) value;
                ComparisonCondition.Operator comp = ((ComparisonCondition)condition).getOperator();
                if (comp == ComparisonCondition.Operator.Equal|| comp == ComparisonCondition.Operator.NotEqual) {
//                  if (valueString.length() <= 32)
//                      sql.append(" AND value in (SELECT id FROM short_str WHERE value='" + XSDBSource.makeStringSQLSafe(valueString) + "')");
  //                else
                      sql.append(" AND value in (SELECT id FROM str WHERE value" + (comp == ComparisonCondition.Operator.NotEqual ? "!":"") + "='" + makeStringSQLSafe(valueString) + "')");
              }
                else {
                    throw new RuntimeException("These comparators have not been implemented for strings yet");
                }
            }
            else if (value instanceof Persistable || value instanceof PersistableList) {
                setValue(value, null, new ValueSetter() {
                   public void setValue(int fieldType,long fieldValue) throws SQLException {
                       sql.append(" AND value_type="+fieldType);
                       sql.append(" AND value="+fieldValue);
                   }
                });
            }
            else {
                throw new RuntimeException("stipulation using value of " + value.getClass() + " has not been implemented yet");
            }
        	
        }
        sql.append(')');
    }

	public Object getFieldValue(LazyPropertyId valueId) throws SQLException {
        ResultSet rs = null;
        PreparedStatement statement = getConnectionObject().stringLookup;
    	statement.setLong(1,Long.parseLong(valueId.subObjectId));
        rs = statement.executeQuery();
        if (valueId instanceof BigNumLookupPair) {
            rs.next();
            Long newBigNum = new Long(rs.getString("value")); // TODO:need to make this work with the different types of numbers
	        rs.close();
            return newBigNum; 
        } 
        else if (valueId instanceof ShortStringLookupPair) {	
	        if (rs.next()) {
	            Object value = convertStringToObject(decodeString(rs.getString("value")));
		        rs.close();
	            return value;
	        }
        }
        rs.close();
        throw new RuntimeException("Unable to find string for id " + valueId);
    }
	
    public List<ObjectId> getReferrers(String objectId) {
    	List<ObjectId> referrerList;
        try {
            PreparedStatement statement = getConnectionObject().referrerLookup;
            ResultSet rs;
        	statement.setLong(1,Long.parseLong(objectId));
        	statement.setLong(2,Long.parseLong(objectId));
        	rs = statement.executeQuery();
            referrerList = new ArrayList();
    		while (rs.next()) 
    			referrerList.add(ObjectId.idForObject(this,Long.toString(rs.getLong(1))));
    		
    		rs.close();
        }
        catch(SQLException e){
        	throw new RuntimeException(e);
        }
		return referrerList;
    	
	}

    static interface ValueSetter {
        void setValue(int fieldType,long fieldValue) throws SQLException;
    }    
    void setValue(Object value, final String referrer, ValueSetter setter) throws SQLException {
        int fieldType = 0;
        long fieldValue = -1;
        if (value instanceof Persistable)
        	value = ((Persistable)value).getId();
        if (value instanceof ObjectId) {
        	((ObjectId)value).persistIfNeeded(new StartAsEmptyPersister() {
        		Object acl = ObjectId.idForObject(DynaObjectDBSource.this, referrer);
        		public Object getAcl() {
        			return acl;
				}
				String newObjectId;
				public void start() throws Exception {
	            	newObjectId = Long.toString(getNextId());
				}
				@Override
				public void recordProperty(String name, Object value)
						throws Exception {
					if (GlobalData.ACCESS_FIELD.equals(name)) {
				    	if (value instanceof ObjectId)
				    		value = ((ObjectId)value).getTarget();
				    	acl = new PersistentAcl((Persistable) value);
					}
					super.recordProperty(name, value);
				}
				@Override
				public void initializeAsList(List values) throws Exception {
        			recordPropertyAddition(newObjectId,GlobalData.CONTAINS_REPEATING_IDENTIFIER,true);
					super.initializeAsList(values);
				}
				public WritableDataSource getSource() {
					return DynaObjectDBSource.this;
				}
				public String getObjectId() {
					return newObjectId;
				}
        	});
        	if (((ObjectId) value).source == this)
        	{
            	fieldType = OBJECT_TYPE;
        		fieldValue = Long.parseLong(((ObjectId) value).subObjectId);
        	}
        	else
        		value = convertObjectToString(value);
        }
        else if (value instanceof Long) {
            fieldType = INTEGER_TYPE;
            fieldValue = ((Long) value).longValue();
        }
        else if (value instanceof Integer) {
            fieldType = INTEGER_TYPE;
            fieldValue = ((Integer) value).intValue();
        }
        else if (value instanceof Boolean) {
            fieldType = BOOLEAN_TYPE;
            fieldValue = (((Boolean)value).booleanValue() ? 1 : 0);
        }
        else if (value == null)
            fieldType = NULL_TYPE;
        else if (value instanceof String) { // this will be done down below
        }
        else {
            value = convertObjectToString(value);
        }
        if (value instanceof String) {
            fieldType = STR;
            fieldValue = createString((String) value);
        }
        setter.setValue(fieldType,fieldValue);
    }
    private void checkAclChangePermission(String objectId) {
    	// we can't change the acl, if the acl has it's own permission that prevents acl modification
    	Acl acl = ObjectId.idForObject(this, objectId).getTarget().getAcl();
    	if (acl instanceof PersistentAcl && 
    			((PersistentAcl)acl).getAclPersistable().getAccessLevel() < 5)
    		throw new ObjectAccessDeniedException(((PersistentAcl)acl).getAclPersistable(), ObjectAccessDeniedException.ACCESS_PERMISSION_LEVEL);

    }
    private void changeAcl(String objectId, Object value) {
    	checkAclChangePermission(objectId);
    	if (value instanceof ObjectId)
    		value = ((ObjectId)value).getTarget();
    	
    	DataSourceHelper.initializeObject(ObjectId.idForObject(this, objectId)).setAcl(new PersistentAcl((Persistable) value));    	
    }
    public void recordPropertyAddition(final String objectId, final String key, Object value) throws SQLException {
    	if (GlobalData.ACCESS_FIELD.equals(key)) 
    		changeAcl(objectId,value);
    	
        setValue(value, objectId, new ValueSetter() {
            public void setValue(int fieldType,long fieldValue) throws SQLException {
            	executeUpdate("INSERT  INTO obj (id,field,value_type,value) values (" + objectId + ",'" + key + "'," + fieldType + ',' + fieldValue + ')');
            } 
        });
    }
    public void recordPropertyChange(final String objectId, final String key, Object value) throws SQLException {
    	if (GlobalData.ACCESS_FIELD.equals(key)) {    		
    		changeAcl(objectId,value);
    	}
		setValue(value, objectId, new ValueSetter() {
            public void setValue(int fieldType,long fieldValue) throws SQLException {
            	
                ResultSet rs = executeQuery("SELECT value_type,value FROM obj where id=" + objectId + " AND field='" + key +"'");
                long stringToRemove = -1;
                if (rs.next())
                	if (rs.getInt("value_type") == STR)
                		stringToRemove =rs.getLong("value"); 
                rs.close();
            	executeUpdate("UPDATE obj SET value_type=" + fieldType + ",value=" + fieldValue + " WHERE id=" + objectId + " AND field='" + key + "'" );
            	if (stringToRemove > -1) {            		
            		// check to make sure it is not used else where
            		rs = executeQuery("SELECT id FROM obj where value=" + stringToRemove + " AND value_type=" + STR);
            		if (!rs.next())
            			executeUpdate("DELETE FROM str WHERE id=" + stringToRemove);
            		rs.close();
            	}
            }
        });
    }


	public void recordPropertyRemoval(String objectId, String name) throws SQLException {
    	if (GlobalData.ACCESS_FIELD.equals(name)) { // if the access field is removed we must go back to inherited acls
        	checkAclChangePermission(objectId);
    		Map<String,Object> objectMap = getObjectMap(objectId);
    		PersistableInitializer initializer = DataSourceHelper.initializeObject(ObjectId.idForObject(this, objectId));
        	if (objectMap.containsKey(GlobalData.PARENT_FIELD)) {
        		if (!(objectMap.get(GlobalData.PARENT_FIELD) instanceof ObjectId))
        			throw new RuntimeException("parent is not an object");
        		initializer.setParent((ObjectId) objectMap.get(GlobalData.PARENT_FIELD));
        	}
        	else {
	        	List<ObjectId> referrers = getReferrers(objectId);
	        	if (!referrers.isEmpty())
	        		initializer.setParent(referrers.get(0));
        	}

    	}
		executeUpdate("DELETE FROM obj WHERE id=" + objectId + " and field='" + name + "'");
	}

	/* List section */
	public void recordListAdd(final String objectId, int index, Object value) throws SQLException {
		throw new UnsupportedOperationException();
	}
	private Object get(String objectId, int index) {
		return ((List) ObjectId.idForObject(this, objectId).getTarget()).get(index);
	}
	private long getRowId(final String objectId, long index) throws SQLException {
    	PreparedStatement selectRowId = getConnectionObject().selectRowId;
    	selectRowId.setLong(1, Long.parseLong(objectId));
    	ResultSet rs = selectRowId.executeQuery();
    	rs.next();
    	for (int i = 0; i < index; i++)
    		if (!rs.next())
    			throw new IndexOutOfBoundsException();
		long rowId = rs.getLong(1);
		rs.close();
		return rowId;
	}
	private void recordListRemoval(final String objectId, final int index) throws SQLException {		
        executeUpdate("DELETE FROM list WHERE row_id=" + getRowId(objectId,index));
	}
	public void recordListReplace(final String objectId, final int index, final Object value) throws SQLException {
    	setValue(value,objectId,new ValueSetter() {
            public void setValue(final int fieldType,final long fieldValue) throws SQLException {                        	
                executeUpdate("UPDATE list SET value_type=" + fieldType + ", value=" + fieldValue + " WHERE row_id=" + getRowId(objectId,index));
            }
        });
	}
	public void recordListAdd(final String objectId, Object value) throws SQLException {		
        setValue(value,objectId, new ValueSetter() {
            public void setValue(int fieldType,long fieldValue) throws SQLException {
                PreparedStatement statement = getConnectionObject().insertRow;
            	statement.setLong(1,Long.parseLong(objectId));
            	statement.setInt(2,fieldType);
            	statement.setLong(3,fieldValue);
                statement.execute();
            }
        });
		
	}
	public void recordListRemoval(final String objectId, Object value) throws SQLException {
        setValue(value,objectId,new ValueSetter() {
            public void setValue(int fieldType,long fieldValue) throws SQLException {
                executeUpdate("DELETE FROM list WHERE list_id=" + objectId + " AND value_type=" + fieldType + " AND value=" + fieldValue);
            }
        });
		
	}
	public void recordListReplace(String objectId, int index, Object value, Stipulation stipulation) throws Exception {
		// TODO Auto-generated method stub
		
	}
	private void shorten(ResultSet rs, long newLength) throws Exception {
    	int i = 0;
    	while (rs.next()) {
    		long rowId = rs.getLong(1);
    		i++;
    		if (i > newLength)
    	        executeUpdate("DELETE FROM list WHERE row_id=" + rowId);
    	}
		rs.close();
	}
	public void truncate(String objectId,long newLength) throws Exception {
    	PreparedStatement selectRowId = getConnectionObject().selectRowId;
    	selectRowId.setLong(1, Long.parseLong(objectId));
    	shorten(selectRowId.executeQuery(),newLength);
	}	
	
	private void truncate(String objectId, long newLength, JsonPath query) throws Exception {
		StringBuffer sql = new StringBuffer("SELECT row_id FROM list WHERE list_id=" + objectId + " AND value IN (SELECT id FROM obj WHERE ");
        addConditionToQuery(query.getWhere(), sql);
        sql.append(')');
        //System.err.println("Sublist query: " + sql);
    	shorten(executeQuery(sql.toString()),newLength);
	}

	

}